package com.linkstec.mot.util;

import java.util.Properties;
import java.util.concurrent.ExecutionException;
import java.util.concurrent.Future;

import org.apache.kafka.clients.producer.KafkaProducer;
import org.apache.kafka.clients.producer.ProducerConfig;
import org.apache.kafka.clients.producer.ProducerRecord;
import org.apache.kafka.clients.producer.RecordMetadata;
import org.apache.kafka.common.KafkaException;

import com.alibaba.fastjson.JSON;

public class KafkaSender {

	private static Properties props;
	
	static 
	{
		props = new Properties();
		props.put(ProducerConfig.BOOTSTRAP_SERVERS_CONFIG, "192.168.56.175:9092,192.168.70.188:9092,192.168.70.189:9092");  
        props.put(ProducerConfig.KEY_SERIALIZER_CLASS_CONFIG, "org.apache.kafka.common.serialization.StringSerializer");
        props.put(ProducerConfig.VALUE_SERIALIZER_CLASS_CONFIG, "org.apache.kafka.common.serialization.StringSerializer");
        props.put(ProducerConfig.ACKS_CONFIG, "all");
//        props.put(ProducerConfig.TIMEOUT_CONFIG, "10000");
//        props.put(ProducerConfig.METADATA_FETCH_TIMEOUT_CONFIG, "10000");
        props.put(ProducerConfig.REQUEST_TIMEOUT_MS_CONFIG, "10000");
        props.put(ProducerConfig.MAX_BLOCK_MS_CONFIG, "10000");
        props.put(ProducerConfig.COMPRESSION_TYPE_CONFIG, "none");
        props.put(ProducerConfig.CLIENT_ID_CONFIG, "producer-test");
        props.put(ProducerConfig.RETRIES_CONFIG, "3");
   
	}
	
	private static ThreadLocal<KafkaProducer<String, String>> threadLocal = new ThreadLocal<KafkaProducer<String, String>>(){
		public KafkaProducer<String, String> initialValue() {
	        return new KafkaProducer<String, String>(props);
	    }
	};
	
	public static KafkaProducer<String, String> getKafkaProducer()
	{
		KafkaProducer<String, String> producer = threadLocal.get();
		if(producer == null)
		{
			producer = new KafkaProducer<String, String>(props);
		}

		return producer;
		
	}
	public static void sendKafkaMessage(String topic, Object kafkamsg) throws Exception
	{
		KafkaProducer<String, String> localProducer = getKafkaProducer();
		String message = JSON.toJSONString(kafkamsg);
		ProducerRecord<String, String> msg = new ProducerRecord<String, String>(topic, message);
		try{
			Future<RecordMetadata> metaDataFuture = localProducer.send(msg);
			metaDataFuture.get();
		}catch(KafkaException e)
		{
			throw e;
		}
		catch(InterruptedException e1)
		{
			throw e1;
		}
		catch(ExecutionException e2)
		{
			throw e2;
		}
	}
	
	public static void close(KafkaProducer<String, String> producer)
	{
		if(producer != null)
		{
			producer.close();
		}
	}
	
	public static void main(String[] args) {
		try {
			String testInfo = "Hello World ";
			KafkaSender.sendKafkaMessage("SERVERINFO", testInfo);
			Long beginTime = System.currentTimeMillis();
			for(int i = 0; i < 1000; i++)
			{
				KafkaSender.sendKafkaMessage("SERVERINFO", testInfo);
			}
			System.out.println("costs " + (System.currentTimeMillis() - beginTime) + " ms.");
		} catch (Exception e) {
			// TODO Auto-generated catch block
			e.printStackTrace();
		}
	}
}
